package com.demoworld;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;

/**
 * Description
 * <p>
 * </p>
 * DATE 2018/9/24.
 *
 * @author liwei5.
 */
public class HbaseDao {
    private static Configuration conf = null;

    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "asxasasx:2181/hbase");
    }

    public static boolean creatTable(String tableName, List<String> families) throws Exception {
        TableName tn = TableName.valueOf(tableName);
        try (Connection connection = ConnectionFactory.createConnection(conf); Table table = connection.getTable(tn);) {
            HTableDescriptor tableDescriptor = new HTableDescriptor(tn);
            for (int i = 0; i <= families.size() - 1; i++) {
                tableDescriptor.addFamily(new HColumnDescriptor(families.get(i)));
            }
            Admin admin = connection.getAdmin();
            if (!admin.tableExists(tn)) {
                admin.createTable(tableDescriptor);
                System.out.println(tableName + " created");
                return true;
            }
        }
        return false;
    }

    public static void put(String tableName, String rowKey, String familyName, String columnName, String value) throws Exception {
        HTable htable = new HTable(conf, Bytes.toBytes(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
        htable.put(put);
    }

    public static Result getResult(String tableName, String rowKey) throws Exception {
        Get get = new Get(Bytes.toBytes(rowKey));
        HTable htable = new HTable(conf, Bytes.toBytes(tableName));
        Result result = htable.get(get);
        return result;
    }

    public static Result getResult(String tableName, String rowKey, String familyName, String columnName) throws Exception {
        Get get = new Get(Bytes.toBytes(rowKey));
        HTable htable = new HTable(conf, Bytes.toBytes(tableName));
        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
        Result result = htable.get(get);
        for (KeyValue k : result.list()) {
            System.out.println(Bytes.toString(k.getFamily()));
            System.out.println(Bytes.toString(k.getQualifier()));
            System.out.println(Bytes.toString(k.getValue()));
            System.out.println(k.getTimestamp());
        }
        return result;
    }

    public static void disableTable(String tableName) throws Exception {
        HBaseAdmin admin = new HBaseAdmin(conf);
        admin.disableTable(tableName);

    }

    public static void dropTable(String tableName) throws Exception {
        HBaseAdmin admin = new HBaseAdmin(conf);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }
}
